package org.flink.api;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @Author: TongRui乀
 * @Date: 2020/11/14 15:29
 * @description： 测试source
 */
public class SourceApiTest {

    private static LocalStreamEnvironment env;

    static {
        env = StreamExecutionEnvironment.createLocalEnvironment();
    }

    public static void test_source_collection() throws Exception {

        DataStreamSource<Integer> stream = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9));

        stream.printToErr();

        env.execute("test source");
    }

    public static void test_source_file() throws Exception {

        DataStreamSource<String> source = env.readTextFile("E:\\code\\java\\flink_demo\\flink_helloworld\\src\\main\\resources\\filesource.txt");

        source.print();

        env.execute();
    }

    public static void test_source_datasource() throws Exception {

        Properties properties = new Properties();

        properties.put("bootstrap.servers", "192.168.199.128:9092");
        properties.put("group.id", "flink_group");

        env.addSource(new FlinkKafkaConsumer<String>("flink_topic", new SimpleStringSchema(),properties));

        // 自定义数据源
        /*env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> sourceContext) throws Exception {

            }

            @Override
            public void cancel() {

            }
        });*/

        env.execute();
    }


    public static void main(String[] args) throws Exception {

//        test_source_collection();
        test_source_file();
    }

}
